package com.shujia.flink.core;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

import java.time.Duration;

public class Demo5EventTime {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /*
         * 水位对齐： 下游task再生成水位时取的是上游所有task最小的水位线
         * 如果数据少，并行度高，会导致水位线对不齐，计算不触发
         * 当将并行度设置为1，水位线就不存在对不齐的情况了
         *
         */
        env.setParallelism(2);

        /**
         * java,1699035731000
         * java,1699035732000
         * java,1699035735000
         * java,1699035733000
         * java,1699035736000
         * java,1699035737000
         * java,1699035740000
         */
        DataStream<String> linesDS = env.socketTextStream("master", 8888);

        //解析数据，取出时间字段
        DataStream<Tuple2<String, Long>> wordAndTsDS = linesDS.map(line -> {
            String[] split = line.split(",");
            String word = split[0];
            //将时间戳转换成long类型
            long ts = Long.parseLong(split[1]);
            return Tuple2.of(word, ts);
        }, Types.TUPLE(Types.STRING, Types.LONG));

        //1、需要告诉flink哪一个字段是时间字段
        //设置时间字段和水位线
        DataStream<Tuple2<String, Long>> assDS = wordAndTsDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        //1、指定水位线等于时间最新一条数据的时间戳，数据不存在乱序的时候使用，如果数据乱序，可能会丢失数据
                        //.<Tuple2<String, Long>>forMonotonousTimestamps()
                        //2、水位线生成方式：最新一条数据的时间戳减去5秒，会导致计算延迟触发
                        .<Tuple2<String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        //指定时间字段
                        .withTimestampAssigner((kv, ts) -> kv.f1)
        );

        /**
         * 实时统计单词的数量，统计最近5秒单词的数量
         */
        DataStream<Tuple2<String, Integer>> kvDS = assDS
                .map(kv -> Tuple2.of(kv.f0, 1),Types.TUPLE(Types.STRING,Types.INT));

        KeyedStream<Tuple2<String, Integer>, String> keyByDS = kvDS.keyBy(kv -> kv.f0);

        //划分窗口
        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowDS = keyByDS
                //滚动的事件时间窗口
                .window(TumblingEventTimeWindows.of(Time.seconds(5)));

        windowDS.sum(1).print();


        env.execute();


    }
}
